package com.tn.rtdata.management;

import java.util.Map;
import java.util.UUID;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.tn.openrpc.util.FactoryException;
import com.tn.openrpc.util.ParametrizedFactory;
import com.tn.rtdata.ControlProcess;
import com.tn.rtdata.DataEvent;
import com.tn.rtdata.DataEventStats;
import com.tn.rtdata.Fill;
import com.tn.rtdata.FillHeartbeat;
import com.tn.rtdata.FillInfo;
import com.tn.rtdata.FillListener;
import com.tn.rtdata.FillException;
import com.tn.rtdata.FillIdentity;

/**
 * A manager service that manages fills.
 */
public class FillManager
{
  private static final Logger LOGGER = LoggerFactory.getLogger(FillManager.class);

  private Map<String, DataEventStats> dataEventStatsCache;
  private ParametrizedFactory<Fill, FillIdentity> fillFactory;
  private Map<FillIdentity, FillInfo> fillInfoCache;
  private Map<FillIdentity, FillHeartbeat> fillHeartbeatCache;
  private long fillTimeout;
  private boolean running;

  /**
   * Creates a new <code>FillManager</code>.
   *
   * @param fillFactory the factory that creates fills when they don't exist.
   */
  public FillManager(
    Map<FillIdentity, FillInfo> fillInfoCache,
    Map<FillIdentity, FillHeartbeat> fillHeartbeatCache,
    Map<String, DataEventStats> dataEventStatsCache,
    ParametrizedFactory<Fill, FillIdentity> fillFactory,
    long fillTimeout
  )
  {
    this.fillInfoCache = fillInfoCache;
    this.fillHeartbeatCache = fillHeartbeatCache;
    this.dataEventStatsCache = dataEventStatsCache;
    this.fillFactory = fillFactory;
    this.fillTimeout = fillTimeout;
  }

  /**
   * Starts the fill manager.
   */
  public void start()
  {
    if (!running)
    {
      LOGGER.info("Starting fill monitor.");

      running = true;

      Thread fillMonitorThread = new Thread(new FillMonitor(fillHeartbeatCache, fillTimeout), "fill-monitor");
      fillMonitorThread.setDaemon(true);
      fillMonitorThread.start();
    }
  }

  /**
   * Stops the fill manager.
   */
  public void stop()
  {
    LOGGER.info("Stopping fill monitor.");
    this.running = false;
  }

  /**
   * Starts the fill with the specified <code>fillIdentity</code> for the <code>fillListener</code>.
   */
  public void startFill(FillIdentity fillIdentity, FillListener fillListener) throws FillException
  {
    try
    {
      Fill fill = fillFactory.newInstance(fillIdentity);
      //noinspection unchecked
      fill.addListener(fillListener);

      if (!fill.isStarted())
      {
        LOGGER.debug("Starting fill for: {}", fillIdentity);
        fill.start();
        fillHeartbeatCache.put(fillIdentity, new FillHeartbeat(fillIdentity, ControlProcess.getId()));
      }
    }
    catch (FactoryException e)
    {
      throw new FillException("An error occurred creating a fill: " + fillIdentity, e);
    }
  }

  /**
   * Stops the fill with the specified <code>fillIdentity</code> for the <code>fillListener</code>.
   */
  public void stopFill(FillIdentity fillIdentity, FillListener fillListener) throws FillException
  {
    try
    {
      Fill fill = fillFactory.newInstance(fillIdentity);
      fill.removeListener(fillListener);
      LOGGER.debug("Removed listener from fill: {}", fillIdentity);

      if (!fill.hasListeners())
      {
        LOGGER.info("Last listener removed, stopping fill: {}", fillIdentity);
        fillHeartbeatCache.remove(fillIdentity);
        fill.stop();
      }
    }
    catch (FactoryException e)
    {
      throw new FillException("An error occurred creating a fill: " + fillIdentity, e);
    }
  }

  /**
   * An implementation of <code>Runnable</code> that heartbeats for each <code>Fill</code> managed by this process.
   */
  private class FillMonitor implements Runnable
  {
    private Map<FillIdentity, FillHeartbeat> fillHeartbeatCache;
    private long fillTimeout;

    /**
     * Creates a new <code>FillMonitor</code>.
     *
     * @param fillHeartbeatCache the cache holding the fill heartbeats.
     * @param fillTimeout        the time allowed between heartbeats before a fill is considered to have failed.
     */
    public FillMonitor(Map<FillIdentity, FillHeartbeat> fillHeartbeatCache, long fillTimeout)
    {
      this.fillHeartbeatCache = fillHeartbeatCache;
      this.fillTimeout = fillTimeout;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void run()
    {
      while (running)
      {
        for (FillIdentity fillIdentity : fillHeartbeatCache.keySet())
        {
          FillHeartbeat fillHeartbeat = fillHeartbeatCache.get(fillIdentity);
          if (fillHeartbeat != null)
          {
            if (ControlProcess.getId().equals(fillHeartbeat.getControlProcessId()))
            {
              LOGGER.debug("Heartbeating fill: {}", fillIdentity);

              //Double check that the fill hasn't been stopped while this loop was running.
              if (fillHeartbeatCache.containsKey(fillIdentity))
              {
                fillHeartbeatCache.put(fillIdentity, new FillHeartbeat(fillHeartbeat));
              }
            }
            else if (System.currentTimeMillis() > fillHeartbeat.getLastHeartbeat().getTime() + fillTimeout)
            {
              LOGGER.warn(
                "Taking control of fill: {} after heartbeat failed from control process: {}.",
                fillIdentity,
                fillHeartbeat.getControlProcessId()
              );
              takeControlOfFill(fillIdentity);
            }
          }
        }

        sleepQuietly(fillTimeout / 4);
      }
    }

    /**
     * Takes control of the fill with the specified <code>fillIdentity</code> running in this process.
     */
    private void takeControlOfFill(FillIdentity fillIdentity)
    {
      try
      {

        @SuppressWarnings({"unchecked"})
        FillInfo<Object, Object> fillInfo = fillInfoCache.remove(fillIdentity);

        if (fillInfo != null)
        {
          Fill fill = fillFactory.newInstance(fillIdentity);

          for (FillListener<Object> fillListener : fillInfo.getListeners())
          {
            fillListener.fillReset(createResetEventId(fillIdentity));
            //noinspection unchecked
            fill.addListener(fillListener);
          }

          fill.start();

          fillHeartbeatCache.put(fillIdentity, new FillHeartbeat(fillIdentity, ControlProcess.getId()));
        }
        else
        {
          LOGGER.warn("Trying to start fill: {} but fill info could not be found.", fillIdentity);
        }
      }
      catch (FactoryException e)
      {
        LOGGER.error("An error occurred trying to get fill: " + fillIdentity, e);
      }
      catch (FillException e)
      {
        LOGGER.error("An error occurred trying to re-start fill: " + fillIdentity, e);
      }
    }

    /**
     * Creates a new data event id, that is used to track data event processing.
     */
    private String createResetEventId(FillIdentity fillIdentity)
    {
      String eventId = UUID.randomUUID().toString();
      dataEventStatsCache.put(eventId, new DataEventStats(eventId, fillIdentity.getName(), DataEvent.Type.RESET));

      return eventId;
    }

    /**
     * Sleeps for the specified <code>millis</code> catching any <code>InterruptedException</code>s.
     */
    private void sleepQuietly(long millis)
    {
      try
      {
        Thread.sleep(millis);
      }
      catch (InterruptedException e)
      {
        LOGGER.debug("Fill monitor thread interrupted.", e);
        //Ignore.
      }
    }
  }
}
